#include "rabbitmq_client.h"

static amqp_connection_state_t conn;
static amqp_basic_properties_t props;
static amqp_channel_t channel = 1;
bool RABBIT_STAUTS = false;
static pthread_t rabbit_pth_id  = -1;

void message_callback(amqp_connection_state_t conn, amqp_envelope_t *envelope) {
    reemoon_ignore("Received message: %s, len: %d\n", (char *)envelope->message.body.bytes, (int)envelope->message.body.len);

    // 识别是否是自己发送的消息
    if (strstr((char *)envelope->message.body.bytes, RABIITMQ_SENDER_ID))
        return ;

    // 不是自己的消息
    int ret = exit_info_JSON((char *)envelope->message.body.bytes);

    if (ret == -2 && envelope->message.body.len > 0) // 不存在cmd信息 ，不是出口信息
    {
        reemoon_write_file("/reemoon/src/img/work_1.txt", (char *)envelope->message.body.bytes, envelope->message.body.len);// 装箱图片信息
        system("base64 -d /reemoon/src/img/work_1.txt > /reemoon/src/img/work_1.png");
        reemoon_set_img_id(1);
        if (system_info->exit_cnt == 2)
        {
            reemoon_write_file("/reemoon/src/img/work_2.txt", (char *)envelope->message.body.bytes, envelope->message.body.len);// 装箱图片信息
            system("base64 -d /reemoon/src/img/work_2.txt > /reemoon/src/img/work_2.png");
            reemoon_set_img_id(2);
        }
    }

    amqp_basic_ack(conn, 1, envelope->delivery_tag, 0);
}


// 接收消息的线程函数
void* rabbitmq_receive_thread(void* arg) {

    // 消息循环
    amqp_envelope_t envelope;
    while (1) {
        amqp_rpc_reply_t consume_reply = amqp_consume_message(conn, &envelope, NULL, 0);
        if (consume_reply.reply_type == AMQP_RESPONSE_NORMAL) {
            message_callback(conn, &envelope);
        } else {
            reemoon_error ("Failed to consume message: %s\n", amqp_error_string2(consume_reply.library_error));
            break;
        }
        usleep(1000);
    }
}

// 启动接收线程
int rabbitmq_start_receive_thread() {
    if (!RABBIT_STAUTS) {
        reemoon_error("RabbitMQ is not initialized.\n");
        return -1;
    }

    if (rabbit_pth_id != -1) {
        reemoon_error("Receive thread already running.\n");
        return -1;
    }

    int ret = pthread_create(&rabbit_pth_id, NULL, rabbitmq_receive_thread, NULL);
    if (ret != 0) {
        reemoon_error("Failed to create receive thread");
        return ret;
    }

    reemoon_ignore ("Receive thread started successfully.\n");
    return 0;
}
int reemoon_msg_send(char *message)
{
    if (!RABBIT_STAUTS) {
        reemoon_error("RabbitMQ is not initialized.\n");
        return -1;
    }

    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.content_type = amqp_cstring_bytes("text/plain");
    props.delivery_mode = 2;  // 持久化消息

    amqp_basic_publish(conn, channel, amqp_cstring_bytes(system_info->rabbitmq_fanout_exchangename), amqp_empty_bytes, 0, 0, &props, amqp_cstring_bytes(message));
    amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Failed to send message: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }

    reemoon_ignore ("Message sent successfully: %s\n", message);
    return 0;
}

// 初始化 RabbitMQ
int rabbitmq_init(void) {
    if (RABBIT_STAUTS) {
        reemoon_warn ("RabbitMQ already initialized.\n");
        return -1;
    }

    amqp_socket_t *socket = NULL;
    amqp_rpc_reply_t reply;

    // 创建连接
    conn = amqp_new_connection();
    socket = amqp_tcp_socket_new(conn);
    if (!socket) {
        reemoon_error("Creating TCP socket failed");
        return 1;
    }

    if (amqp_socket_open(socket, system_info->server_ip, 5672)) {
        reemoon_ignore ("Opening TCP socket failed");
        return 1;
    }

    // 登录
    reply = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, system_info->rabbitmq_username, system_info->rabbitmq_password);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Login failed: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }

    // 打开通道
    amqp_channel_open(conn, channel);
    reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Opening channel failed: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }

    // 声明交换机
    amqp_exchange_declare(conn, channel, amqp_cstring_bytes(system_info->rabbitmq_fanout_exchangename), amqp_cstring_bytes("fanout"), 0, 0, 0, 0, amqp_empty_table);
    reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Declaring exchange failed: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }

    // 声明队列
    amqp_bytes_t queue_name;
    amqp_queue_declare_ok_t *queue_declare_result = amqp_queue_declare(conn, channel, amqp_cstring_bytes(system_info->rabbitmq_simple_queuename), 0, 0, 0, 0, amqp_empty_table);
    reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Declaring queue failed: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }
        if (amqp_get_rpc_reply(conn).reply_type == AMQP_RESPONSE_NORMAL) {
        queue_name = amqp_bytes_malloc_dup(queue_declare_result->queue);
        if (queue_name.bytes == NULL) {
            reemoon_error ("Out of memory while copying queue name\n");
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            RABBIT_STAUTS = false;
            return 1;
        }
        reemoon_ignore ("Queue declared successfully. Queue name: %.*s\n", (int)queue_name.len, (char *)queue_name.bytes);
    } else {
        reemoon_error ("Failed to declare queue: %s\n", amqp_error_string2(amqp_get_rpc_reply(conn).library_error));
        amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
        amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(conn);
        RABBIT_STAUTS = false;
        return 1;
    }

    // 绑定队列到交换机
    amqp_queue_bind(conn, channel, queue_name, amqp_cstring_bytes(system_info->rabbitmq_fanout_exchangename), amqp_empty_bytes, amqp_empty_table);
    reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        reemoon_error("Binding queue to exchange failed: %s\n", amqp_error_string2(reply.library_error));
        return 1;
    }
    
    // 开始消费消息
    amqp_basic_consume(conn, channel, queue_name, amqp_cstring_bytes("consumer_tag"), 0, 0, 0, amqp_empty_table);
    amqp_rpc_reply_t consume_reply = amqp_get_rpc_reply(conn);
    if (consume_reply.reply_type == AMQP_RESPONSE_NORMAL) {
    } else {
        reemoon_error("Failed to start consuming messages: %s\n", amqp_error_string2(consume_reply.library_error));
        amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
        amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(conn);
        return 1;
    }

    RABBIT_STAUTS = 1;
    reemoon_info("RabbitMQ initialized successfully.\n");
    rabbitmq_start_receive_thread();
    return 0;
}


int rabbitmq_deinit(void) 
{
    if (rabbit_pth_id != -1) {
        pthread_cancel(rabbit_pth_id);
        rabbit_pth_id = -1;
    }
    if (!RABBIT_STAUTS) return -1;
    // 关闭通道和连接
    amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
    amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
    amqp_destroy_connection(conn);
    memset (conn, 0, sizeof(conn));
    reemoon_info ("rabbitmq deinit\n");
    RABBIT_STAUTS = false;
}
